package com.test;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.IterativeStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.junit.Test;

public class NumberStreamDemo {
    @Test
    public void test() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<Long> someIntegers = env.generateSequence(0, 10);
        IterativeStream<Long> iteration = someIntegers.iterate();
        DataStream<Long> minusOne = iteration
                .map((MapFunction<Long, Long>) value -> value - 1);

        DataStream<Long> stillGreaterThanZero = minusOne
                .filter(v -> v > 0);
        iteration.closeWith(stillGreaterThanZero);

        DataStream<Long> lessThanZero = minusOne
                .filter(v -> v <= 0);
        iteration.closeWith(lessThanZero);

        stillGreaterThanZero.print();
        env.execute();
    }
}
